package com.demo.rocketmq.producer.ha;

import com.demo.rocketmq.constants.TopicConstants;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@Service
public class Dledger {

    @Resource(name = "rocketMQTemplate")
    private RocketMQTemplate rocketMqTemplate;

    @Resource
    private RocketMQProperties rocketMQProperties;

    /**
     * 发送消息时，master宕机，消息不再继续发送
     */
    public void send() {
        try {
            for (int i = 0; i < 10000; i++) {
                String msg = String.format("第%d条消息,日期：%s", i, new Date());
                rocketMqTemplate.syncSend(TopicConstants.DLEAGER_TOPIC, msg);
                log.info(msg);
            }
        } catch (Exception e) {
            log.warn("broker异常");
        }
        log.info("发送消息完毕");
    }

    /**
     * master宕机后，故障转移后从新的master broker消费消息
     * @throws MQClientException
     * @throws InterruptedException
     */
    public void read() throws MQClientException, InterruptedException {
        log.info("开始消费");
        Map<String, AtomicInteger> map = new ConcurrentHashMap<>();
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("dledger_consume", true, StringUtils.EMPTY);
        consumer.setNamesrvAddr(rocketMQProperties.getNameServer());
        consumer.setConsumeThreadMin(1);
        consumer.setConsumeThreadMax(1);
        // 设置每次批量消费的消息数量。
        // 注意，必须在配合broker的maxTransferCountOnMessageInMemory使用，maxTransferCountOnMessageInMemory默认值是32。
        // 就算消费端设置的值大于32，broker还是会返回给消费端32条消息
        consumer.setConsumeMessageBatchMaxSize(100);
        consumer.setMessageModel(MessageModel.CLUSTERING);
        consumer.subscribe(TopicConstants.DLEAGER_TOPIC, "*");
        consumer.setMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            log.info("批量拉取消息数量：{}", msgs.size());
            msgs.forEach(msg -> {
                try {
                    TimeUnit.MILLISECONDS.sleep(5);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }

                String text = new String(msg.getBody());
                AtomicInteger atomicInteger = map.computeIfAbsent(text, (key) -> new AtomicInteger());
                atomicInteger.incrementAndGet();
                log.info("dledger_consume消费者消费的消息：{}", text);
            });
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        TimeUnit.MINUTES.sleep(2);
        map.forEach((key, value) -> {
            if (value.get() > 1) {
                log.info("重复消费消费：{}，消费次数：{}", key, value.get());
            }
        });
        log.info("消费完毕，消费消息数量：{}", map.size());
    }
}
